package com.idanchuang.support.etcache.pubsub;

import com.idanchuang.support.etcache.util.InetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.management.ManagementFactory;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author yjy
 * Created at 2022/3/28 3:22 下午
 */
public abstract class AbstractRemoteEventPublisher implements RemoteEventPublisher {

    private static final Logger logger = LoggerFactory.getLogger(AbstractRemoteEventPublisher.class);

    private final ExecutorService asyncWorker;
    protected volatile boolean shutdown = false;

    /** 当前实例数据 */
    public static final byte[] LOCAL_INSTANCE_BYTES;

    static {
        // 初始化当前实例数据
        String ip = InetUtil.getLocalIp();
        String processId = ManagementFactory.getRuntimeMXBean().getName().split("@")[0];
        byte[] ipBytes = ip.getBytes(StandardCharsets.UTF_8);
        byte[] processIdBytes = processId.getBytes(StandardCharsets.UTF_8);
        // 前15位为ip信息, 后7位为进程id信息
        LOCAL_INSTANCE_BYTES = new byte[22];
        System.arraycopy(ipBytes, 0, LOCAL_INSTANCE_BYTES, 15 - ipBytes.length, ipBytes.length);
        System.arraycopy(processIdBytes, 0, LOCAL_INSTANCE_BYTES, 22 - processIdBytes.length, processIdBytes.length);
    }

    {
        asyncWorker = new ThreadPoolExecutor(1, 1,
                1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(10),
                r -> new Thread(r, "et-cache-worker"),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    /**
     * 发布缓存变更消息
     * @param remoteKeyBytes 事件数据
     */
    @Override
    public void publish(byte[] remoteKeyBytes) {
        if (shutdown) {
            return;
        }
        asyncWorker.execute(() -> {
            try {
                // 事件数据前20个字节是触发者的实例信息(ip+进程id)
                byte[] eventBytes = new byte[remoteKeyBytes.length + LOCAL_INSTANCE_BYTES.length];
                System.arraycopy(LOCAL_INSTANCE_BYTES, 0, eventBytes, 0, LOCAL_INSTANCE_BYTES.length);
                System.arraycopy(remoteKeyBytes, 0, eventBytes, LOCAL_INSTANCE_BYTES.length, remoteKeyBytes.length);
                doPublish(eventBytes);
            } catch (Exception e) {
                logger.error("doPublisher event failed", e);
            }
        });
    }

    /**
     * 发布缓存变更消息
     * @param remoteKeyBytes 事件数据
     * @throws Exception e...
     */
    public abstract void doPublish(byte[] remoteKeyBytes) throws Exception;

    /**
     * 销毁
     */
    @Override
    public void destroy() {
        if (!this.shutdown) {
            this.shutdown = true;
        }
        asyncWorker.shutdown();
    }

    public boolean isShutdown() {
        return shutdown;
    }
}
